package com.offcn.bigdata.datastream.source

import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala._

/**
 * @Auther: BigData-LGW
 * @ClassName: UserDefineSource
 * @Date: 2020/12/15 21:55
 * @功能描述: $FunctionDescription
 * @Version:1.0
 */
object UserDefineSource {
    def main(args: Array[String]): Unit = {
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        val ret = env.addSource(new MySourceFunction())
        ret.print().setParallelism(2)
        env.execute(s"${UserDefineSource.getClass.getSimpleName}")
    }
class MySourceFunction extends SourceFunction[Int]{
    override def run(ctx: SourceFunction.SourceContext[Int]): Unit = {
        var num =1
        while (true){
            println("上游sourceFunction生产的数据："+num)
            ctx.collect(num)
            num+=1
            Thread.sleep(500)
        }
    }

    override def cancel(): Unit = {

    }
}
}
